package com.csx;

import com.csx.base.*;
import com.csx.base.meta.CheckDataException;
import com.csx.base.meta.OdsColumn;
import com.csx.base.utils.CRCUtils;
import com.csx.base.utils.ThreadUtil;
import com.xxl.job.core.util.DateUtil;
import com.yh.csx.bsf.core.db.DbHelper;
import com.yh.csx.bsf.core.thread.ThreadPool;
import com.yh.csx.bsf.core.util.*;
import lombok.Data;
import lombok.val;
import lombok.var;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 数据校验程序
 * 
 * @author Robin.Wang
 */
public class CheckDbDataTask {
	@Data
	public static class CheckInfo{
		private String table;
		private String result;
		private long errorCount;
		private StringBuilder message= new StringBuilder();
		private Date from;
		private Date end;
		@Override
		public String toString(){
			return "--------------------------------------------------------------------------------<br/>"
					+table +":" + DateUtil.formatDateTime(from) +"->"+ (end==null?"": DateUtil.formatDateTime(end)) +" 结果:"+ StringUtils.nullToEmpty(result)  +"<br/>"
					+ message.toString();
		}
	}
	/**
	 * 策略类
	 */
	StragtegyProvider stragtegyProvider= Context.StragtegyProvider;

	TidbAdapter syncer= Context.OdsSyncAdapter;

	private Map<String,CheckInfo> checkInfos = new ConcurrentHashMap<>();
	private Map<String,CheckInfo> lastCheckInfos = new ConcurrentHashMap<>();
	public List<CheckInfo> getLastCheckInfos(){
		return new ArrayList<>(lastCheckInfos.values());
	}
	public List<CheckInfo> getCheckInfos(){
		return new ArrayList<>(checkInfos.values());
	}

	public void run(){
		while (!ThreadPool.System.isShutdown()){
				try{
					//获取mysql 和 适配库的表关系映射
					HashMap<String,List<String>> mysqlToAdapter = new HashMap<>();
					var mysqlTables = Context.getMysqlTables().values();
					var tidbTables = Context.getTidbTables().values();
					for(var mysqlTable1:mysqlTables){
						var mysqlTable = mysqlTable1.toLowerCase();

						var targetTable = stragtegyProvider.getTargetTable(stragtegyProvider.parseDb(mysqlTable),stragtegyProvider.parseTable(mysqlTable));
						if(tidbTables.contains(targetTable)){
							if(!mysqlToAdapter.containsKey(targetTable)) {
								var list =new ArrayList<String>();list.add(mysqlTable);
								mysqlToAdapter.put(targetTable,list);
							}else {
								var mysqlList = mysqlToAdapter.get(targetTable);
								if(!mysqlList.contains(mysqlTable)) {
									mysqlToAdapter.get(targetTable).add(mysqlTable);
								}
							}
						}
					}
					var tableInfos = mysqlToAdapter.entrySet().stream().collect(Collectors.toList());
					ThreadUtils.parallelFor("并行数据检查",Config.getCheckDataParallelCount(),tableInfos,(tableInfo)->{
						var find = checkInfos.get(tableInfo.getKey());
						CheckInfo checkInfo = (find==null?new CheckInfo():find);
						checkInfo.table = tableInfo.getKey();
						checkInfo.from=new Date();
						checkInfo.message = new StringBuilder();
						try {
							//表结构校验
							if(Config.getCheckIndexEnabled()) {
								checkFiled(checkInfo, tableInfo.getKey(), tableInfo.getValue());
							}
							//校验数据内容
							checkContent(checkInfo,tableInfo.getKey(),tableInfo.getValue());
							//校验更新内容
							checkUpdate(checkInfo,tableInfo.getKey(),tableInfo.getValue());
							//校验新增和删除
							if((","+Config.getCheckDataCreateTimeBlackTable().toLowerCase()+",").indexOf(","+tableInfo.getKey().toLowerCase()+",")<=-1) {
								checkCreateTimeCount(checkInfo, tableInfo.getKey(), tableInfo.getValue());
							}
							checkInfo.result="success";
							checkInfo.errorCount=0;
						}catch (Exception exp){
							checkInfo.result="error";
							checkInfo.errorCount++;
							//2次报错，才视为错误,tidb 怀疑有短暂的不一致问题
							//if(checkInfo.errorCount>1) {
								LogUtils.error(this.getClass(), Config.ProjectName, "表" + tableInfo.getKey() + "校验异常,exp:"+exp.getMessage()+",查看报告 http://" + NetworkUtils.getIpAddress() + ":" + PropertyUtils.getPropertyCache("server.port", 0) + "/");
								if(!(exp instanceof CheckDataException)){
									LogUtils.error(this.getClass(), Config.ProjectName, "表" + tableInfo.getKey() + "校验异常,堆栈:"+ExceptionUtils.trace2String(exp)+",查看报告 http://" + NetworkUtils.getIpAddress() + ":" + PropertyUtils.getPropertyCache("server.port", 0) + "/");
								}
							//}
						}
						checkInfo.end=new Date();
						if (!checkInfos.containsKey(checkInfo.table)) {
							synchronized (this) {
								if (!checkInfos.containsKey(checkInfo.table)) {
									checkInfos.put(checkInfo.table, checkInfo);
								}
							}
						}
					});
					lastCheckInfos= new HashMap<>();
					lastCheckInfos.putAll(checkInfos);
					LogUtils.info(this.getClass(),Config.ProjectName,"后台数据校验完毕");

				}catch (Exception e){
					LogUtils.error(this.getClass(),Config.ProjectName,"后台数据校验任务异常",e);
				}
				ThreadUtil.sleep(Config.getIncrementCheckDataTimeSpan());//*60
			}
	}

	private void checkUpdate(CheckInfo checkInfo, String adapterTable, List<String> mysqlTables){
		checkInfo.message.append("检查更新");
		int perCount = 500/mysqlTables.size();
		Map<String,Date> ids = new HashMap<>();
		var checkDateTime =new Date(System.currentTimeMillis() - Config.getIncrementCheckDataUpdateTimeTimeSpan());
		var sql = "select id,update_time from #{datatable} where update_time<'"+DateUtil.formatDateTime(checkDateTime)+"' limit 0,"+perCount;
		for(var mysql:mysqlTables){
			DbHelper.call(stragtegyProvider.getDS(stragtegyProvider.parseDb(mysql)),(c)->{
				if (!MysqlUtils.existsTable(stragtegyProvider.parseDb(mysql), stragtegyProvider.parseTable(mysql))) {
					return;
				}

				var rs = c.executeList(sql.replace("#{datatable}", mysql),null);
				for(var r:rs){
					ids.put(ConvertUtils.convert(r.get("id"),String.class),ConvertUtils.convert(r.get("update_time"),Date.class));
				}
			});
		}
		checkInfo.message.append("条数:"+ids.size());
		if(syncer.existsTable(adapterTable)){
			List<String> errorIds=new ArrayList<>();
			var rs = syncer.selectUpdate(adapterTable,ids.keySet().stream().collect(Collectors.toList()));
			for(var r:rs.entrySet()){
				var from = ids.get(r.getKey());
				var to = r.getValue();
				if(from==null||to==null||to.getTime()<from.getTime()){
					errorIds.add(r.getKey());
				}
			}
			if(errorIds.size()>0) {
				String msg = adapterTable + " id:" + String.join(",", errorIds) + "更新时间异常";
				checkInfo.message.append("[error]" + msg + "[/error]\r\n");
				throw new CheckDataException(msg);
			}
		}
		checkInfo.message.append("[success]更新校验完成[/success]\r\n");
	}


	private void checkCreateTimeCount(CheckInfo checkInfo, String adapterTable, List<String> mysqlTables){
		int error=0;
		StringBuilder sb = new StringBuilder();
		while(true) {
			try {
				sb.append("检查create_time条数\r\n");
				var checkDateTime = new Date(System.currentTimeMillis() - Config.getIncrementCheckDataTimeSpan());
				final String sql = "select count(0) as num from #{datatable} where create_time<='" + DateUtil.formatDateTime(checkDateTime) + "' ";
				long mysqlcount = 0;
				long adaptercount = 0;
				for (var mysql : mysqlTables) {
					long count = DbHelper.get(stragtegyProvider.getDS(stragtegyProvider.parseDb(mysql)), (c) -> {
						if (!MysqlUtils.existsTable(stragtegyProvider.parseDb(mysql), stragtegyProvider.parseTable(mysql))) {
							return 0L;
						}
						return ConvertUtils.convert(c.executeScalar(sql.replace("#{datatable}", mysql), null), long.class);
					});
					sb.append(mysql + "数量:" + count + ",\r\n");
					mysqlcount = mysqlcount + count;
				}
				if (syncer.existsTable(adapterTable)) {
					adaptercount = syncer.selectCount(adapterTable, checkDateTime);
				}
				if (Math.abs(adaptercount - mysqlcount) < Config.getCheckDataCreateTimeAllowDiffCountSpan(adapterTable)) {
					var msg = mysqlTables + "总数量:" + mysqlcount + "->" + adapterTable + "总数量:" + adaptercount;
					if (adaptercount == mysqlcount) {
						sb.append(msg + "\r\n");
					} else {
						sb.append("[warn]" + msg + "[/warn]\r\n");
					}
				} else {
					String msg = adapterTable + " 检查create_time条数异常:" + mysqlcount + "->" + adaptercount;
					sb.append("[error]" + msg + "[/error]\r\n");
					throw new CheckDataException(msg);
				}
				sb.append("[success]条数校验完成[/success]\r\n");
				checkInfo.message.append(sb.toString());
				break;
			} catch (Exception e) {
				error++;
				if (error >= Config.getCheckDataCreateTimeMaxErrorCount()) {
					checkInfo.message.append(sb.toString()+" 检查次数:"+error);
					throw e;
				}
				sb = new StringBuilder();
				ThreadUtil.sleep(10000);
			}
		}
	}

	private void checkFiled(CheckInfo checkInfo,String adapterTable,List<String> mysqlTables){
		checkInfo.message.append("检查表结构字段标准");
		Map<String, OdsColumn> columns = new HashMap<>();
		val sourceDb = stragtegyProvider.parseDb(mysqlTables.get(0));
		val sourceTable = stragtegyProvider.parseTable(mysqlTables.get(0));
		var tableColumns = MysqlUtils.getTableColumns(sourceDb,sourceTable);
		for(var c: tableColumns){
			columns.put(c.getColumnName(),c);
		}
		if(columns.get("id")==null||columns.get("update_time")==null||columns.get("create_time")==null){
			var msg="表:"+adapterTable+"没有id或update_time或create_time字段";
			checkInfo.message.append("[error]" + msg + "[/error]\r\n");
			LogUtils.error(this.getClass(),Config.ProjectName,msg);
			throw new CheckDataException(msg);
		}
		if(!MysqlUtils.existsIndex(sourceDb,sourceTable,"create_time")||
				!MysqlUtils.existsIndex(sourceDb,sourceTable,"update_time"))
		{
			var msg="表:"+adapterTable+"没有update_time或create_time 索引";
			checkInfo.message.append("[error]" + msg + "[/error]\r\n");
			LogUtils.error(this.getClass(),Config.ProjectName,msg);
			throw new CheckDataException(msg);
		}
		checkInfo.message.append("[success]表结构校验完成"+"[/success]\r\n");
	}



	private void checkContent(CheckInfo checkInfo,String adapterTable,List<String> mysqlTables){
		checkInfo.message.append("检查内容");
		String sqlTemplate = "(select *,id as odsId from #{datatable} order by id desc  limit 20)" + "	union "
				+ "	(select *,id as odsId from #{datatable} order by id asc  limit 5)" + "	union all"
				+ "	(select *,id as odsId from #{datatable} where id> (select * from ((select ((max(id)-min(id))/2) from #{datatable}) b)) order by id desc limit 20)";
		List<Map<String, Object>> check = new ArrayList<>();
		for(var mysql:mysqlTables){
			DbHelper.call(stragtegyProvider.getDS(stragtegyProvider.parseDb(mysql)),(c)->{
				if(!MysqlUtils.existsTable(stragtegyProvider.parseDb(mysql),stragtegyProvider.parseTable(mysql))){return;}
				var rs = c.executeList(sqlTemplate.replace("#{datatable}",mysql),null);
				for(var r:rs){
					check.add(r);
				}
			});
		}
		Map<String, OdsColumn> columns = new HashMap<>();
		var tableColumns = MysqlUtils.getTableColumns(stragtegyProvider.parseDb(mysqlTables.get(0)),stragtegyProvider.parseTable(mysqlTables.get(0)));
		for(var c: tableColumns){
			//标准列内容校验
			String fullColumn = adapterTable+"."+c.getColumnName();
			if(Config.getCheckDataBlackColumns().contains(","+fullColumn+",")) {
				String msg ="跳过列数据检查:"+fullColumn;
				checkInfo.message.append("[warn]"+msg+"[/warn]\r\n");
				LogUtils.info(this.getClass(),Config.ProjectName,"表:"+adapterTable+msg);
			}
			else {
				columns.put(c.getColumnName(), c);
			}
		}
		checkInfo.message.append("检查条数:"+check.size());
		for(var o:check){
			//2分钟内的更新数据,不需要校验,避免数据未同步
			Date fromDate = ConvertUtils.convert(o.get("update_time"),Date.class);
			if(fromDate ==null || System.currentTimeMillis()-fromDate.getTime()<1000*60*2)
			{
				continue;
			}
			//时间戳不相同的数据,不需要校验
			Map<String,Object> d=null;
			try {
				 d = syncer.getById(adapterTable, o.get("odsId"));
				if (d == null || d.size() == 0) {
					//新增的数据不存在
					continue;
				}
			}catch (Exception exp){
				var msg = "表" + adapterTable + " id:" + o.get("odsId") + "tidb获取数据出错";
				checkInfo.message.append("[error]" + msg + "[/error]\r\n");
				throw new CheckDataException(msg);
			}
		    try {
				CRCUtils.checkEqualData(o.get("update_time"), d.get("update_time"), columns.get("update_time"));
			}catch (Exception e){
		    	var from = CommonUtils.objToStr(o.get("update_time")); var to= CommonUtils.objToStr(d.get("update_time"));
				LogUtils.info(this.getClass(),Config.ProjectName,"表:"+adapterTable+" id:"+o.get("odsId")+"发现update_time校验"+from+"->"+to+"不一致,跳过此行内容验证");
				continue;
			}
		    //校验相同字段内容
			for (var key : o.keySet()) {
				if (d.containsKey(key)) {
					try {
						//标准列内容校验
						if (columns.containsKey(key)) {
							CRCUtils.checkEqualData(o.get(key), d.get(key), columns.get(key));
						}
					} catch (Exception e) {
						var msg = "表" + adapterTable + " id:" + o.get("odsId") + ",列" + key + "数据[内容]校验不一致:" + e.getMessage();
						checkInfo.message.append("[error]" + msg + "[/error]\r\n");
						throw new CheckDataException(msg);
					}
				}
			}

		}
		checkInfo.message.append("[success]内容校验完成"+"[/success]\r\n");
	}

}
